package com.mayikt.canal.kafka.client;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.mayikt.utils.RedisUtil;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@Component
public class MembetKafkaConsumer {
    @Autowired
    private RedisUtil redisUtil;

    @KafkaListener(topics = "mayikt-topic")
    public void receive(ConsumerRecord<?, ?> consumer) {
        System.out.println("topic名称:" + consumer.topic() + ",key:" +
                consumer.key() + "," +
                "分区位置:" + consumer.partition()
                + ", 下标" + consumer.offset() + "," + consumer.value());
        String json = (String) consumer.value();
        JSONObject jsonObject = JSONObject.parseObject(json);
        String type = jsonObject.getString("type");
        String pkNames = jsonObject.getJSONArray("pkNames").getString(0);
        JSONArray data = jsonObject.getJSONArray("data");
        String table = jsonObject.getString("table");
        String database = jsonObject.getString("database");
        for (int i = 0; i < data.size(); i++) {
            JSONObject dataObject = data.getJSONObject(i);
            String key = database + ":" + table + ":" + dataObject.getString(pkNames);
            switch (type) {
                case "UPDATE":
                case "INSERT":
                    redisUtil.setString(key, dataObject.toJSONString());
                    break;
                case "DELETE":
                    redisUtil.delKey(key);
                    break;
            }
        }
    }
}
